package com.atguigu.gmall.common.config;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.common.entity.GmallCorrelationData;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

/**
 * @Description 消息发送确认
 * <p>
 * ConfirmCallback  只确认消息是否正确到达 Exchange 中
 * ReturnCallback   消息没有正确到达队列时触发回调，如果正确到达队列不执行
 * <p>
 * 1. 如果消息没有到exchange,则confirm回调,ack=false
 * 2. 如果消息到达exchange,则confirm回调,ack=true
 * 3. exchange到queue成功,则不回调return
 * 4. exchange到queue失败,则回调return
 *
 */
@Component
@Slf4j
public class MQProducerAckConfig implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnCallback {

    private static final String CACHE_RABBIT = "cacheRabbit";

    @Resource
    private RabbitTemplate rabbitTemplate;

    @Resource
    private RedisTemplate redisTemplate;

    @PostConstruct
    public void init(){
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
    }

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (ack){
            System.out.println("交换机应答成功：" + cause);
        } else {
            System.out.println("交换机应答失败：" + cause);
            retrySend(correlationData);
        }
    }

    private void retrySend(CorrelationData correlationData) {
        GmallCorrelationData gmallCorrelationData = (GmallCorrelationData) correlationData;
        int retryCount = gmallCorrelationData.getRetryCount();
        if (retryCount < 3){
            gmallCorrelationData.setRetryCount(++retryCount);
            redisTemplate.opsForHash().put(CACHE_RABBIT,gmallCorrelationData.getId()
                    ,JSONObject.toJSONString(gmallCorrelationData));
            log.info("重新发送消息：" + JSONObject.toJSONString(gmallCorrelationData));
            rabbitTemplate.convertAndSend(gmallCorrelationData.getExchange(),
                    gmallCorrelationData.getRoutingKey(),
                    gmallCorrelationData.getMessage(),
                    gmallCorrelationData);
        } else {
            log.info("重新发送消息完毕：" + JSONObject.toJSONString(gmallCorrelationData));
        }
    }

    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {

        Object uuid = message.getMessageProperties().
                getHeader("spring_returned_message_correlation");
        String data = (String) redisTemplate.opsForHash().get(CACHE_RABBIT, uuid);
        GmallCorrelationData gmallCorrelationData = JSONObject.parseObject(data, GmallCorrelationData.class);
        retrySend(gmallCorrelationData);
    }
}
